package spider.zookeeper;

import org.apache.zookeeper.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import spider.zookeeper.util.StringUtils;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

/**
 * BaseZKClient
 *
 * @author Archx[archx@foxmail.com]
 * @date 2017/4/7
 */
public abstract class BaseZKClient {

    private static final Logger logger = LoggerFactory.getLogger(BaseZKClient.class);

    protected ZooKeeper zooKeeper;

    /**
     * 服务器连接信息
     */
    private String servers;

    /**
     * Session 超时毫秒
     */
    private int timeout;

    /**
     * 状态计数器
     */
    private CountDownLatch connectLatch = new CountDownLatch(1);
    private CountDownLatch disconnectLatch = new CountDownLatch(1);

    /**
     * 建立连接
     */
    public void establish() {
        try {
            zooKeeper = new ZooKeeper(servers, timeout, new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                        connectLatch.countDown();
                        logger.info("> ZK server connected");
                    } else if (watchedEvent.getState() == Event.KeeperState.Disconnected) {
                        logger.info("> ZK server disconnected");
                        disconnectLatch.countDown();
                    } else if (watchedEvent.getState() == Event.KeeperState.Expired) {
                        // 过期关闭重连
                        logger.info("> ZK session expired");
                        close();
                        retry();
                        logger.info("> ZK server retry connected");
                    }
                }
            });

            // 等待建立连接
            connectLatch.await();

            // 执行方法
            execute();

            // 阻塞直到连接断开
            disconnectLatch.await();

        } catch (IOException e) {
            logger.error(StringUtils.printStackTrace(e));
        } catch (InterruptedException e) {
            logger.error(StringUtils.printStackTrace(e));
            logger.info("> ZK server connection interrupted");
        } finally {
            logger.info("> ZK server connection loss");
        }
    }

    /**
     * 执行方法
     */
    public abstract void execute();

    /**
     * 创建节点
     *
     * @param path       路径
     * @param data       数据
     * @param createMode 模式
     */
    public void createNode(String path, byte[] data, CreateMode createMode) {
        zooKeeper.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, createMode, new AsyncCallback.StringCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx, String name) {
                switch (KeeperException.Code.get(rc)) {
                    case CONNECTIONLOSS:
                        logger.info("> Connection loss: [ path = {} ]", path);
                        break;
                    case OK:
                        logger.info("> Node created: [ path = {} ]", path);
                        break;
                    case NODEEXISTS:
                        logger.info("> Node already registered: [ path = {} ]", path);
                        break;
                    default:
                        logger.error("> Something went wrong: \n" + KeeperException
                                .create(KeeperException.Code.get(rc), path));
                }
            }
        }, data);
    }

    /**
     * 移除节点
     *
     * @param path 路径
     */
    public void removeNode(String path) {
        try {
            zooKeeper.delete(path, -1);
        } catch (Exception e) {
            logger.error(StringUtils.printStackTrace(e));
        }
    }

    /**
     * 获取节点数据
     *
     * @param path 路径
     * @return
     */
    public String getNodeData(String path) {
        try {
            return new String(zooKeeper.getData(path, false, null));
        } catch (Exception e) {
            logger.error(StringUtils.printStackTrace(e));
        }
        return null;
    }

    /**
     * 添加观察者
     * <p>
     * 通过反应器执行结果判断是否重复执行节点操作
     *
     * @param action  节点操作
     * @param handler 反应处理器
     */
    public void addWatcher(final NodeAction action, final WatcherReactor handler) {
        Watcher watcher = new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                boolean watchNext = handler.handle(watchedEvent);
                if (watchNext) {
                    runNodeAction(action, this);
                }
            }
        };
        runNodeAction(action, watcher);
    }

    /**
     * 执行节点操作
     *
     * @param action  节点操作
     * @param watcher 观察者
     */
    private void runNodeAction(NodeAction action, Watcher watcher) {
        try {
            action.doAction(watcher);
        } catch (InterruptedException | KeeperException e) {
            logger.error(StringUtils.printStackTrace(e));
        }
    }

    private void close() {
        if (zooKeeper != null) {
            try {
                zooKeeper.close();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                zooKeeper = null;
            }
        }
    }

    private void retry() {
        if (zooKeeper == null) {
            synchronized (BaseZKClient.class) {
                if (zooKeeper == null) {
                    connectLatch = new CountDownLatch(1);
                    disconnectLatch = new CountDownLatch(1);
                    establish();
                }
            }
        }
    }

    public void setServers(String servers) {
        this.servers = servers;
    }

    public void setTimeout(int timeout) {
        this.timeout = timeout;
    }

}
